package com.mq.config;

import cn.hutool.json.JSONUtil;
import com.mq.domain.SysDept;
import com.mq.domain.SysUser;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.concurrent.TimeUnit;

/**
 * 消费消息
 * @author gxd
 * @since 2025/6/8
 */
@Slf4j
@Component
public class ReceiveData {

    @Autowired
    private RedisTemplate redisTemplate;

    // 消费用户消息
    @RabbitListener(queues = "sys.user.queue", ackMode = "MANUAL")
    public void receiveUserData(Message message,  Channel channel) throws IOException {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        String msgStr = new String(message.getBody());
        SysUser user = JSONUtil.toBean(msgStr, SysUser.class);
        // 组id + 唯一标识字段
        String lockKey = "user_lock:" + user.getGroupId() + user.getPhone();

        boolean lockAcquired = false;
        try {
            // redis setnx上锁
            lockAcquired = redisTemplate.opsForValue().setIfAbsent(lockKey, "给当前消息加锁", 10, TimeUnit.MINUTES);

            // 上锁失败，说明key已存在，消息已被消费过，手动标记消费
            if (Boolean.FALSE.equals(lockAcquired)) {
                // 业务处理消息
                log.warn("该消息已被消费过不被消费了, id={}", user.getId());
                // 手动标记已消费
                channel.basicAck(deliveryTag, false);
                return;
            }

            // 上锁成功，存入key，消费消息，手动标记消费
            // 业务处理消息
            log.info("未被消费进行消费, 放行id={}", user.getId());
            // 手动标记消费消息
            channel.basicAck(deliveryTag, false);
        } catch (Exception e) {
            // 报错，重新消费
            // 手动标记未消费
            channel.basicNack(deliveryTag, false, true);
            log.error("消费失败id={}", user.getId());
        }
    }

    // 消费部门消息
    @RabbitListener(queues = "sys.dept.queue", ackMode = "MANUAL")
    public void receiveDeptData(Message message, Channel channel) throws IOException {
        // 获取数据
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        String msgStr = new String(message.getBody());
        SysDept dept = JSONUtil.toBean(msgStr, SysDept.class);

        String lockKey = "dept_lock:" + dept.getGroupId() + dept.getId();
        boolean lockAcquired = false;

        try {
            lockAcquired = redisTemplate.opsForValue().setIfAbsent(lockKey, "给该消息加锁", 10, TimeUnit.MINUTES);

            if (Boolean.FALSE.equals(lockAcquired)) {
                log.warn("该消息已被消费过不被消费了，id={}", dept.getId());
                channel.basicAck(deliveryTag, false);
                return;
            }

            log.info("未被消费进行消费，id={}", dept.getId());
            channel.basicAck(deliveryTag, false);
        } catch (Exception e) {
            // 手动标记未消费
            channel.basicNack(deliveryTag, false, true);
            log.error("实体类：{}, 消费失败", dept);
        }
    }
    // mq使用场景：数据同步、预警发送 websocket 和 短信提醒
    // 单体(不重复)消息：直接消费
    // 并发消息： redisson锁 //
    // 并发消息 和 多条相同（重复）消息：生产者定义组id，例如：预警id， key为预警id + 业务唯一标识， redisson锁 + false的时候手动标记消费
    // 经常异常的（也就是无法正常消费的消息） 可以选择存储到数据库，手动处理异常消息数据，手动标记已消费
    // 或者把这条记录存入到另一个专门处理异常消息数据的队列中，当前管道手动标记已消费，异常队列管道还能消费
    // 消息丢失，生产者多发几份，防止消息丢失。
}
